package com.fmdj.snm.task;

import cn.hutool.core.map.MapUtil;
import com.fmdj.common.exception.GlobalException;
import com.fmdj.snm.entry.MessageEntity;
import com.fmdj.snm.entry.MessageRefEntity;
import com.fmdj.snm.service.MessageService;
import com.rabbitmq.client.*;
import com.rabbitmq.client.impl.AMQBasicProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
@Slf4j
public class MessageTask {

    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    /**
     * 同步发送私有消息
     * @param identity 交换机
     * @param userId 接收者id
     * @param ttl 过期时间
     * @param entity 发送的实体内容
     */
    public void sendPrivateMessage(String identity, long userId, Integer ttl, MessageEntity entity) {
        //发送消息之前先把消息写进数据库
        String id = messageService.insertMessage(entity); //返回的是messageId

        //发送消息
        String exchangeName = identity +"_private";
        String queueName = "queue_" + userId;
        String routingKey = userId + "";


        try (
                Connection connection = factory.newConnection(); //创建连接
                Channel channel = connection.createChannel()
        ) {
            //声明交换机(定向)
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            //放在队列中的参数(可以不设置)
            HashMap params = new HashMap<>();
            channel.queueDeclareNoWait(queueName, true, false, false, params); //这里的参数是可选的 可为空

            //存放在属性的headers中
            HashMap map = new HashMap<>();
            map.put("messageId", id); //发送的消息的id
            //设置发送消息的属性
            AMQP.BasicProperties properties;
            if (ttl!=null&&ttl>0) {  //添加过期时间
                    properties = new AMQP.BasicProperties().builder()
                            .contentEncoding("UTF-8")
                            .headers(map)
                            .expiration(ttl.toString())
                            .build();
            }else {
                properties = new AMQP.BasicProperties().builder()
                        .contentEncoding("UTF-8")
                        .headers(map)
                        .build();
            }

            //发送消息     entity.getMsg()消息字符串可以从respond中获取 respond.getBody();
            channel.basicPublish(exchangeName, routingKey, properties, entity.getMsg().getBytes()); //从对象中获取发送的msg 因为要发送字节 所以要getBytes()

            log.debug("发送消息成功");
        } catch (Exception e) {
            log.error("执行错误",e);
            throw new GlobalException("向RaabbitMq发送消息失败");
        }
    }

    //异步发送私有消息
    @Async
    public void sendPrivateMessageAsync(String identity, long userId, Integer ttl, MessageEntity entity) {
        sendPrivateMessage(identity, userId, ttl, entity);
    }

    /**
     * 获取账单消息
     * @param identity
     * @param userId
     * @return 返回接收响应体中的msg
     */
    public String receiveBillMessage(String identity, long userId) {
        String exchangeName = identity +"_private";
        String queueName = "queue_" + userId;
        String routingKey = userId + "";

        try (
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel()
        ) {
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            channel.queueDeclareNoWait(queueName, true, false, false, null); //这里的参数是可选的 可为空
            channel.basicQos(0,1,true); //限制提取消息数量(防止内存溢出)为1条 对全局生效
            channel.queueBind(queueName, exchangeName, routingKey); //通过routingKey经队列绑定到交换机上(发送消息不用做这一步,接收消息需要做绑定这一步)
            GetResponse response = channel.basicGet(queueName, false);//获取消息并且取消自动确认

            if (response!=null) {
                AMQP.BasicProperties properties = response.getProps();
                Map<String, Object> map = properties.getHeaders();
                String messageId = MapUtil.getStr(map, "messageId");
                String msg = new String(response.getBody());//取出消息体

                log.debug("成功从RabbitMq中接收消息");
                //成功接收消息后把数据持久化到数据库中
                MessageRefEntity entity = new MessageRefEntity();
                entity.setMessageId(messageId);//发送消息id
                entity.setReceiverId(userId); //接收者id
                entity.setReceiverIdentity(identity); //交换机
                entity.setReadFlag(true); //设置已读
                entity.setLastFlag(true);//设置为是最新消息
                messageService.insertMessageRef(entity);

                long deliveryTag = response.getEnvelope().getDeliveryTag();
                channel.basicAck(deliveryTag, false); //手动确认收到一条消息

                return msg; //返回接收到的消息
            }
        } catch (Exception e) {
            log.error("执行错误", e);
            throw new GlobalException("从rabbitMq中接收消息失败");
        }

        //没有接收到消息
        return null;
    }
}
